package org.example.com.atguigu.day03;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.junit.Test;
import scala.Tuple2;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Arrays;

public class $01_Test1 implements Serializable {
    @Test
    public void test1() {
        SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("com.atguigu.day01.$01_RddCreate");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> rdd1 = sc.parallelize(Arrays.asList("a 1", "b 2", "b 3", "c 1", "d 1", "e 5"));
        JavaPairRDD<String, Integer> rdd2 = rdd1.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<>(s.split(" ")[0], Integer.parseInt(s.split(" ")[1]));
            }
        });
        // rdd2 [(a,1), (b,2), (b,3), (c,1), (d,1), (e,5)]
        rdd2.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                Connection connection = null;
                PreparedStatement statement = null;
                try {
                    // 1 创建mysql连接
                    connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/spark_test_db", "root", "root");
                    // 2 创建statement对象
                    statement = connection.prepareStatement("insert into test2 values(?,?)");
                    // 3 参数赋值
                    statement.setString(1, stringIntegerTuple2._1);
                    statement.setInt(2, stringIntegerTuple2._2);
                    // 4 插入数据
                    statement.executeUpdate();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 5 关闭连接
                    if (statement != null)
                        statement.close();
                    if (connection != null)
                        connection.close();
                }

            }
        });
    }
}
